#include "netlib/net/socket/tcp_connection.h"

#include <utility>

// #include <functional>

#include "netlib/base/logger.h"
// #include "buffer.h"
#include "netlib/net/event/event_loop.h"

netlib::net::TcpConnection::TcpConnection(
    EventLoop* loop, int fd, const InetAddress& local, const InetAddress& peer, std::string name) :
    loop_(loop),
    name_(std::move(name)), kSocket(fd), kLocalAddr(local), kPeerAddr(peer), channel_(loop, fd) {
	channel_.SetReadCallback([this](auto&& PH1) { HandleRead(std::forward<decltype(PH1)>(PH1)); });
	// channel_.setCloseCallback(
	//     std::bind(&TcpConnection::handleClose, this));
	channel_.SetWriteCallback([this] { HandleWrite(); });
	channel_.SetErrorCallback([this] { HandleError(); });
}

netlib::net::TcpConnection::~TcpConnection() = default;

void netlib::net::TcpConnection::Established() {
	loop_->AssertInLoopThread();
	assert(state_ == kConnecting);
	state_ = kConnected;

	channel_.EnableReading();
	conn_cb_(shared_from_this());
}

// void netlib::net::TcpConnection::connectEstablished() {
// 	loop_->assertInLoopThread();
// 	assert(state_ == kConnecting);
// 	state_ = kConnected;

// 	channel_.enableReading();
// 	conn_cb_(shared_from_this());
// }

void netlib::net::TcpConnection::Send(const std::string& msg) {
	if (state_ == kConnected) {
		if (loop_->IsInLoopThread()) {
			SendInLoop(msg);
		} else {
			loop_->QueueInLoop([this, msg] { SendInLoop(msg); });
		}
	}
}

void netlib::net::TcpConnection::Shutdown() {
	if (state_ == kConnected) {
		state_ = kDisconnecting;
		LOG_INFO << "connection " << name_ << " shutdown";
		// channel_.disableWriting();
		loop_->RunInLoop([this] { ShutdownInLoop(); });
	}
}

void netlib::net::TcpConnection::Destroyed() {
	loop_->AssertInLoopThread();
	assert(state_ == kDisconnected);
	// state_ = kDisconnected;
	// channel_.disableAll();
	assert(channel_.IsNoneEvent());
	channel_.Remove();
}

void netlib::net::TcpConnection::HandleRead(Timestamp time) {
	int save_errno = 0;
	int len = input_buffer_.ReadFd(kSocket.Fd(), &save_errno);
	if (len > 0) {
		msg_cb_(shared_from_this(), &input_buffer_, time);
	} else if (len < 0) {
		errno = save_errno;
		LOG_SYSERR << "TcpConnection::handleRead()";
		HandleError();
	} else {
		HandleClose();
	}
}

void netlib::net::TcpConnection::HandleClose() {
	loop_->AssertInLoopThread();
	// LOG_INFO << stateToString() << "handleClose called by
	// thread[" << CurrentThread::tid() << "]";
	if (state_ == kDisconnected) {
		return;
	}
	assert(state_ == kConnected || state_ == kDisconnecting);
	channel_.DisableAll();
	state_ = kDisconnected;
	assert(channel_.IsNoneEvent());
	// if (state_ == kConnected) state_ = kDisconnecting;
	TcpConnectionPtr this_conn(shared_from_this());
	conn_cb_(this_conn);
	close_cb_(this_conn);
}

void netlib::net::TcpConnection::HandleWrite() {
	if (channel_.IsWriting()) {
		int len = static_cast<int>(
		    ::write(kSocket.Fd(), output_buffer_.Peek(), output_buffer_.ReadableBytes()));
		if (len > 0) {
			output_buffer_.Retrieve(len);
			if (output_buffer_.ReadableBytes() == 0) {
				channel_.DisableWriting();
				if (state_ == kDisconnecting) {
					ShutdownInLoop();
				}
			}
		} else {
			LOG_SYSERR << "TcpConncetion::handleWrite()";
			channel_.DisableWriting();
		}
	} else {
		LOG_INFO << "connection [" << name_ << "] is " << StateToString() << ", no more writing";
		channel_.DisableWriting();
	}
}

const char* netlib::net::TcpConnection::StateToString() const {
	switch (state_) {
	case kInit:
		return "Init";
	case kDisconnected:
		return "Disconnected";
	case kConnecting:
		return "Connecting";
	case kConnected:
		return "Connected";
	case kDisconnecting:
		return "Disconnecting";
	default:
		return "unknown state";
	}
}

void netlib::net::TcpConnection::HandleError() {
	int err = sockets::GetSocketError(kSocket.Fd());
	LOG_ERROR << "TcpConnection::handleError [" << name_ << "] - SO_ERROR = " << err << " "
	          << StrerrorTl(err);
}

void netlib::net::TcpConnection::SendInLoop(const std::string& msg) {
	loop_->AssertInLoopThread();
	int len = 0;
	int msg_size = static_cast<int>(msg.size());
	if (!channel_.IsWriting() && output_buffer_.ReadableBytes() == 0) {
		// if (state_ != kConnected) {
		// 	LOG_WARN << "connection " << name_ << " is "
		// 	         << stateToString() << " , unable to write ";
		// }
		len = static_cast<int>(::write(kSocket.Fd(), msg.c_str(), msg.size()));
		if (len < 0) {
			len = 0;
			if (errno != EAGAIN) {
				LOG_SYSERR << "TcpConnection::sendInLoop()";
			}
		}
	}
	if (len < msg_size) {
		output_buffer_.Append(&msg[len], msg_size - len);
		if (!channel_.IsWriting()) {
			channel_.EnableWriting();
		}
	}
}

void netlib::net::TcpConnection::ShutdownInLoop() {
	loop_->AssertInLoopThread();
	if (!channel_.IsWriting()) {
		sockets::ShutdownWrite(kSocket.Fd());
	} else {
		LOG_INFO << "channel is writing";
	}
}
